package com.captjack.kafka.application;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * 启动入口
 *
 * @author Capt Jack
 * @date 2018/7/4
 */
@SpringBootApplication
@RestController
@EnableKafka
public class KafkaApplication {

    /**
     *
     */
    private static final String TEST_TOPIC = "hypt_fence";

    /**
     * 版本
     */
    @Value(value = "${startVersion}")
    private String startVersion;

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @GetMapping(value = "/")
    public String hello() {
        kafkaTemplate.send(TEST_TOPIC, String.valueOf(System.currentTimeMillis()));
        return this.startVersion;
    }

    @KafkaListener(topics = {TEST_TOPIC})
    public void consumer(ConsumerRecord<String, String> record) {
        try {
            System.out.println("topic:" + record.topic());
            System.out.println("key:" + record.key());
            System.out.println("offset:" + record.offset());
            System.out.println("value:" + record.value());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * @param args 参数
     */
    public static void main(String[] args) {
        SpringApplication.run(KafkaApplication.class, args);
    }

}
